package service

import (
	"encoding/json"
	"fmt"
	"log"
	"time"

	"fabric-sdk/utils"

	"github.com/IBM/sarama"
	cb "github.com/hyperledger/fabric-protos-go/common"

	// "github.com/hyperledger/fabric-sdk-go/pkg/client/channel"
	"github.com/hyperledger/fabric-sdk-go/pkg/client/event"
	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/fab"
)

type EventService struct {
	ChaincodeID string
	// Client      *channel.Client
	EventClient []*event.Client
}

var prevTimeStamp int64 = 0

func (t *EventService) StartEventService(kafkaBroker string) {
	if t.EventClient == nil {
		fmt.Println("failed to get event client")
		return
	}
	for i := 0; i < len(t.EventClient); i++ {
		_, notifier, err := t.EventClient[i].RegisterBlockEvent()
		if err != nil {
			fmt.Println("failed to register block event")
		}
		go t.ReceiveEventResult(notifier, kafkaBroker)
	}
}

// func regitserEvent(client *channel.Client, chaincodeID, eventID string) (fab.Registration, <-chan *fab.CCEvent) {

// 	reg, notifier, err := client.RegisterChaincodeEvent(chaincodeID, eventID)
// 	if err != nil {
// 		fmt.Println("注册链码事件失败: ", err)
// 	} else {
// 		fmt.Println("注册链码事件成功!")
// 	}
// 	return reg, notifier
// }

func (t *EventService) ReceiveEventResult(notifier <-chan *fab.BlockEvent, kafkaBroker string) error {
	fmt.Println("wait event result ... ")
	config := sarama.NewConfig()
	config.Producer.Return.Successes = true
	config.Producer.Timeout = 5 * time.Second
	p, err := sarama.NewSyncProducer([]string{kafkaBroker}, config)
	if err != nil {
		log.Printf("sarama.NewSyncProducer err, message=%s \n", err)
		return err
	}
	defer p.Close()
	for {
		select {
		case ccEvent := <-notifier:
			if err := ExtractObjIDsFromBlockToKafka(ccEvent.Block, &p); err != nil {
				return err
			}
		}
	}
}

func ExtractObjIDsFromBlockToKafka(Block *cb.Block, p *sarama.SyncProducer) error {
	timeStart := time.Now().UnixNano()
	objIDsInBlock, err := utils.ExtractObjIDsFromBlock(Block)
	if err != nil {
		return err
	}

	messages, err := json.Marshal(objIDsInBlock)
	if err != nil {
		return err
	}

	msg := &sarama.ProducerMessage{
		Topic: "block",
		Value: sarama.ByteEncoder(messages),
	}
	_, _, err = (*p).SendMessage(msg)
	if err != nil {
		log.Printf("send message(%s) err=%s \n", messages, err)
	}
	timeEnd := time.Now().UnixNano()
	objNum := len(objIDsInBlock.IDs)
	fmt.Printf("接收到链码事件: blocknum -> %d, 解析区块花费时间: %fms, obj num: %d, TPS: %f\n", Block.Header.Number, (float64)(timeEnd-timeStart)/1e6, objNum, (float64)(objNum)*1e9/(float64)(timeStart-prevTimeStamp))
	prevTimeStamp = timeStart
	return nil
}
